package com.copc.service.impl;

import cn.hutool.core.lang.UUID;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.copc.copc.entity.MerchantConfig;
import com.copc.copc.service.IMerchantConfigService;
import com.copc.mq.config.ConfirmMessageCallback;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static java.lang.System.out;

/**
 * @author yuanbin
 * @version 1.0
 */
@Service
public class MsgServiceImpl {

    private Map<String, AmqpTemplate> templates = new ConcurrentHashMap<>();

    @Autowired
    @Qualifier("prototypeAmqpTemplate")
    private ObjectProvider<AmqpTemplate> amqpTemplateProvider;

    @Resource
    private IMerchantConfigService merchantConfigService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    public void setConfirmCallback(RabbitTemplate rabbitTemplate){
        rabbitTemplate.setConfirmCallback(new ConfirmMessageCallback());
    }

    private AmqpTemplate getTemplate(String accountId) {
        QueryWrapper<MerchantConfig> merchantConfigWrapper = new QueryWrapper();
        merchantConfigWrapper.eq("account_id",accountId);//启用的商户配置
        MerchantConfig merchantConfig = merchantConfigService.getOne(merchantConfigWrapper);
        String uri = merchantConfig.getAmqpUri();
        return amqpTemplateProvider.getObject(uri);
    }

    /**
     * 发送普通消息到消费队列
     * @param queue
     * @param msg
     */
    public void sendMsg(String queue,Object msg) {
        String id = UUID.randomUUID().toString();
        // 该对象将在回调时携带到回调方法中
        CorrelationData correlationData = new CorrelationData(id);
        rabbitTemplate.convertAndSend(queue,msg,correlationData);
    }

    public void sendMessage(String accountId, String queue, Object message) {
        AmqpTemplate template = templates.computeIfAbsent(accountId, this::getTemplate);
        template.convertAndSend(queue, message);
    }


    /**
     * 一个队列绑定多个消费者
     * @param exchangeName
     * @param msg
     */
    public void sendManyConsumer(String exchangeName, String msg) {
        String id = UUID.randomUUID().toString();
        // 该对象将在回调时携带到回调方法中
        CorrelationData correlationData = new CorrelationData(id);
        rabbitTemplate.convertAndSend(exchangeName,"",msg,correlationData);
    }

    /**
     * 路由key
     * @param exchangeName
     * @param routeKey
     * @param msg
     */
    public void sendToManyConsumerByRoute(String exchangeName, String routeKey, String msg) {
        rabbitTemplate.convertAndSend(exchangeName,routeKey,msg);
    }

    /**
     * 发送参数类型是object
     * @param queue
     * @param mapMsg
     */
    public void sendObjectConsumer(String queue, Map<String, Object> mapMsg) {
        rabbitTemplate.convertAndSend(queue,mapMsg);
    }

    /**
     * 发送延时消息到mq
     * @param exchange 死信交换机
     * @param routeKey 路由key
     * @param data 发送数据
     * @param delayTime 过期时间，单位毫秒
     */
    public void sendDelayMsgToMQ(String exchange, String routeKey, String data,int delayTime) {
        rabbitTemplate.convertAndSend(exchange, routeKey, data, message -> {
            message.getMessageProperties().setExpiration(delayTime + "");
            return message;
        });
    }

}
